package com.example.ra.service.impl;


import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.example.ra.domain.dto.MarketingDTO;
import com.example.ra.domain.po.CustRecords;
import com.example.ra.domain.po.CustRecordsR;
import com.example.ra.domain.vo.MarketingRecordVO;
import com.example.ra.mapper.CustRecordsMapper;
import com.example.ra.service.CustRecordsService;
import jakarta.annotation.Resource;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Comparator;
import java.util.List;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Service
public class CustRecordsServiceImpl extends ServiceImpl<CustRecordsMapper, CustRecords>
    implements CustRecordsService {

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private CustRecordsMapper custRecordsMapper;


    @Autowired
    private RabbitTemplate rabbitTemplate;

    private final String EXCHANGE_NAME = "delay.direct";

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void insertCustRecord(MarketingDTO mtd) {
        String custId = mtd.getCustId();
        // 投送到MQ延迟队列
        rabbitTemplate.convertAndSend("delay.direct", mtd.getSendType(), mtd,msg -> {
            msg.getMessageProperties().setDelayLong(mtd.getDelay());
            return msg;
        });

        CustRecords custRecords = new CustRecords();
        custRecords.setCustId(custId);
        custRecords.setSendFlag(1);
        custRecords.setSendTime(LocalDateTime.now());

        // 1. 写入数据库
        custRecordsMapper.insert(custRecords);

        // 2. 同步到 Redis
        String key = "marketing:records:" + custId;
        if (Boolean.TRUE.equals(redisTemplate.hasKey(key))) {
            String member = JSON.toJSONString(custRecords);
            long score = custRecords.getSendTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
            redisTemplate.opsForZSet().add(key, member, score);
            redisTemplate.expire(key, 1, TimeUnit.DAYS); // 刷新过期时间
        }
    }

    @Override
    public MarketingRecordVO get30(String custId) {
        String key = "marketing:records:" + custId;
        long now = System.currentTimeMillis();
        long thirtyDaysAgo = now - 30L * 24 * 60 * 60 * 1000;

        // 从 Redis 查询
        Set<String> cachedData = redisTemplate.opsForZSet().rangeByScore(key, thirtyDaysAgo, now);
        if (cachedData != null && !cachedData.isEmpty()) {
            List<CustRecordsR> records = cachedData.stream()
                    .map(r -> JSON.parseObject(r, CustRecordsR.class))
                    .sorted(Comparator.comparing(CustRecordsR::getSendTime).reversed())
                    .collect(Collectors.toList());
            return buildResponse(records);
        }

        // 缓存未命中，查询数据库
        List<CustRecordsR> dbRecords = custRecordsMapper.get30(custId);

        // 将数据同步到 Redis
        if (!dbRecords.isEmpty()) {
            redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
                dbRecords.forEach(record -> {
                    String member = JSON.toJSONString(record);
                    long score = record.getSendTime().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
                    connection.zAdd(key.getBytes(), score, member.getBytes());
                });
                return null;
            });
            redisTemplate.expire(key, 1, TimeUnit.DAYS); // 设置过期时间
        }

        return buildResponse(dbRecords);

    }

    private MarketingRecordVO buildResponse(List<CustRecordsR> records) {
        MarketingRecordVO vo = new MarketingRecordVO();
        vo.setDateList(records);
        return vo;
    }
}




